package com.kgc.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.flume.FlumeUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}


/**
  * Spark Streaming整合Flume操作：Poll风格的推方法
  */
object FlumePullWordCount {
  def main(args: Array[String]) {
    if (args.length < 2) {
      System.err.println(
        "Usage: FlumePushWordCount <host> <port>")
      System.exit(1)
    }

    val Array(hostname, port) = args
    val sparkConf = new SparkConf().setAppName("FlumePullWordCount").setMaster("local[2]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    //获取flume数据
    val flumeStream = FlumeUtils.createPollingStream(ssc, hostname, port.toInt, StorageLevel.MEMORY_ONLY_SER_2)
    flumeStream.map(x =>  new String(x.event.getBody.array()).trim).flatMap(_.split(" "))
      .map(word => (word, 1)) // 每个单词映射成一个pair
      .reduceByKey(_ + _) // 根据每个key进行累加
      .print() // 打印前10个数据
    ssc.start()
    ssc.awaitTermination()
  }

}
